package com.atguigu.hbase;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

/**
 * Created by 黄凯 on 2023/6/30 0030 19:50
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 */
public class HBase04_DML {

    public static Connection connection = Hbase02_SingletonConnection.getConnection();

    public static void main(String[] args) throws IOException {
//        testPutData("", "emp" , "1001" , "f1" , "name" , "zhangsan");
//        testDeleteData("" , "emp" , "1001" ,"f1" , "name");
//        testGetData("" , "stu" , "1001");
//        testScanData("" , "stu" , "1002" , "1003!");

//        testScanDataWithFilter("" , "stu" , "f2" , "age" , "28");
//        testScanDataWithOneFilter("" , "stu" , "f1" , "sex" , "m");
        testScanDataWithMultiFilter("" , "stu");

        connection.close();
    }

    /**
     * Hbase是否支持基于非rowkey列数据的过滤扫描?
     *   支持，但是慎用， 因为要全表扫描。
     *   scan的时候通过Filter来进行过滤扫描
     *
     * select id , name, addr from table where id >= ?   基于rowkey
     *
     * select id, name , addr from table where name = ? or addr = ?   基于非rowkey
     *
     *
     * 需求:
     *   过滤年龄大于等于28岁的数据
     *
     *   过滤性别等于m的数据
     *
     * 1004 : f1 : addr : tianjing
     * 1004 : f1 : name : wangwu
     * 1004 : f1 : sex : f
     * 1004 : f2 : age : 28
     *
     */

    public static void testScanDataWithMultiFilter(String namespaceName ,String tableName  ) throws IOException {
        TableName tn = TableName.valueOf(namespaceName, tableName);
        //获取Table对象
        Table table = connection.getTable(tn);
        //扫描操作
        Scan scan = new Scan();

        //设置Filter
        //过滤年龄大于等于28岁的数据
        SingleColumnValueFilter ageFilter = new SingleColumnValueFilter(
                Bytes.toBytes("f2"),
                Bytes.toBytes("age"),
                CompareOperator.GREATER_OR_EQUAL,
                Bytes.toBytes("28")
        );
        ageFilter.setFilterIfMissing(true);

        //过滤性别等于m的数据
        SingleColumnValueFilter sexFilter = new SingleColumnValueFilter(
                Bytes.toBytes("f1"),
                Bytes.toBytes("sex"),
                CompareOperator.EQUAL,
                Bytes.toBytes("f")
        );
        sexFilter.setFilterIfMissing(true);

        // 通过FilterList封装多个条件
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL /*and*/, ageFilter, sexFilter);
        //FilterListWithAND filterListWithAND = new FilterListWithAND(Arrays.asList(ageFilter , sexFilter));

        scan.setFilter(filterList);

        ResultScanner resultScanner = table.getScanner(scan);
        for (Result result : resultScanner) {
            //处理每行数
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {

                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String cl = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println(rowkey + " : " + cf + " : " + cl + " : " + value );

            }

            System.out.println("-----------------------------------------");
        }

        table.close();
    }


    public static void testScanDataWithOneFilter(String namespaceName ,String tableName , String cfFilter , String clFilter , String valueFilter ) throws IOException {
        TableName tn = TableName.valueOf(namespaceName, tableName);
        //获取Table对象
        Table table = connection.getTable(tn);
        //扫描操作
        Scan scan = new Scan();

        //设置Filter
        //过滤年龄大于等于28岁的数据
        SingleColumnValueFilter ageFilter = new SingleColumnValueFilter(
                Bytes.toBytes(cfFilter),
                Bytes.toBytes(clFilter),
                CompareOperator.GREATER_OR_EQUAL,
                Bytes.toBytes(valueFilter)
        );
        //scan.setFilter(ageFilter);

        //过滤性别等于m的数据
        SingleColumnValueFilter sexFilter = new SingleColumnValueFilter(
                Bytes.toBytes(cfFilter),
                Bytes.toBytes(clFilter),
                CompareOperator.EQUAL,
                Bytes.toBytes(valueFilter)
        );

        //我们可以设置filter.setFilterIfMissing(true);如果为true，当这一列不存在时，不会返回，如果为false，当这一列不存在时，会返回所有的列信息
        sexFilter.setFilterIfMissing(true);
        scan.setFilter(sexFilter);

        ResultScanner resultScanner = table.getScanner(scan);
        for (Result result : resultScanner) {
            //处理每行数
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {

                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String cl = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println(rowkey + " : " + cf + " : " + cl + " : " + value );

            }

            System.out.println("-----------------------------------------");
        }

        table.close();
    }


    /**
     * 扫描 - scan
     *
     * Shell : scan 'namespaceName:tableName' , {STARTROW=>'' , STOPROW=>''}
     *
     * 1001 : f1 : addr : beijing
     * 1001 : f1 : name : zhangsan
     * 1001 : f1 : sex : m
     * 1001 : f2 : age : 25
     * -----------------------------------------
     * 1002 : f1 : addr : beijing
     * 1002 : f1 : name : zhangsan
     * 1002 : f1 : sex : f
     * 1002 : f2 : age : 25
     * -----------------------------------------
     * 1003 : f1 : addr : shanghai
     * 1003 : f1 : name : lisi
     * 1003 : f1 : sex : m
     * 1003 : f2 : age : 30
     * -----------------------------------------
     * 10031 : f1 : addr : shenzhen
     * 10031 : f1 : name : zhaoliu
     * 10031 : f1 : sex : m
     * 10031 : f2 : age : 30
     * -----------------------------------------
     * 1004 : f1 : addr : tianjing
     * 1004 : f1 : name : wangwu
     * 1004 : f1 : sex : f
     * 1004 : f2 : age : 28
     * -----------------------------------------
     * 1005 : f1 : addr : beijing
     * 1005 : f1 : name : tianqi
     * 1005 : f2 : age : 29
     * -----------------------------------------
     */
    public static void testScanData(String namespaceName, String tableName, String startRow , String stopRow) throws IOException {
        TableName tn = TableName.valueOf(namespaceName, tableName);
        //获取Table对象
        Table table = connection.getTable(tn);

        //扫描操作
        Scan scan = new Scan();
        //设置范围
        scan.withStartRow(Bytes.toBytes(startRow)).withStopRow(Bytes.toBytes(stopRow));

        ResultScanner resultScanner = table.getScanner(scan);
        for (Result result : resultScanner) {
            //处理每行数
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {

                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String cl = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println(rowkey + " : " + cf + " : " + cl + " : " + value );

            }

            System.out.println("-----------------------------------------");
        }

        table.close();
    }


    /**
     * 查询 - get
     *
     * Shell : get 'namespaceName:tableName' , 'rk'
     *
     *
     * 1001 : f1 : addr : beijing
     * 1001 : f1 : name : zhangsan
     * 1001 : f1 : sex : m
     * 1001 : f2 : age : 25
     */
    public static void testGetData(String namespaceName, String tableName ,String rk ) throws IOException {
        TableName tn = TableName.valueOf(namespaceName, tableName);
        //获取Table对象
        Table table = connection.getTable(tn);

        //get操作
        Get get = new Get(Bytes.toBytes(rk));
        Result result = table.get(get);
        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {
            /*
            String rowKey = Bytes.toString(cell.getRowArray());
            String cf = Bytes.toString(cell.getFamilyArray());
            String cl = Bytes.toString(cell.getQualifierArray());
            String value = Bytes.toString(cell.getValueArray()) ;
            System.out.println(rowKey + " : " + cf + " : " + cl + " : " + value );

             */

            String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
            String cf = Bytes.toString(CellUtil.cloneFamily(cell));
            String cl = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            System.out.println(rowkey + " : " + cf + " : " + cl + " : " + value );

        }

        table.close();
    }



    /**
     * 删除 - delete / deleteall
     *
     * Shell:
     *       delete 'namespaceName:tableName' , 'rk' , 'cf:cl'   (Delete)
     *       deleteall 'namespaceName:tableName' , 'rk' , 'cf:cl' (DeleteColumn)
     *       deleteall 'namespaceName:tableName' , 'rk'  (DeleteFamily)
     */
    public static void testDeleteData(String namespaceName, String tableName ,String rk , String cf, String cl) throws IOException {
        TableName tn = TableName.valueOf(namespaceName, tableName);
        //获取Table对象
        Table table = connection.getTable(tn);

        //删除操作
        Delete delete = new Delete(Bytes.toBytes(rk));
        //Delete
//        delete.addColumn(Bytes.toBytes(cf) , Bytes.toBytes(cl)) ;

        //DeleteColumn
//        delete.addColumns(Bytes.toBytes(cf), Bytes.toBytes(cl));

        //DeleteFamily
        delete.addFamily(Bytes.toBytes(cf));


        table.delete(delete) ;
        table.close();
    }



    /**
     *  插入/修改 - put
     *
     *  Shell: put 'namespaceName:tableName', 'rk' , 'cf:cl' , 'value'
     */
    public static void testPutData(String namespaceName , String tableName , String rk , String cf , String cl , String value) throws IOException {
        TableName tn = TableName.valueOf(namespaceName , tableName) ;
        //获取Table对象
        Table table = connection.getTable(tn);
        //put操作
        //创建Put对象
        Put put = new Put(Bytes.toBytes(rk));
        //设置其他信息
        put.addColumn(Bytes.toBytes(cf) , Bytes.toBytes(cl) , Bytes.toBytes(value)) ;

        table.put(put);

        table.close();
    }


}
